//! # 数据库存储适配器
//!
//! 提供基于SQL数据库的消息持久化实现，支持PostgreSQL、MySQL等主流数据库

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use sqlx::{Pool, Postgres, MySql, Row};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use uuid::Uuid;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;
use crate::persistence::{
    MessageRecord, MessageStatus, MessageStorage, QueryCriteria, 
    OrderBy, SortOrder, StorageStats, StorageTransaction, RetryDelayStrategy
};

/// 数据库类型枚举
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum DatabaseType {
    PostgreSQL,
    MySQL,
    SQLite,
}

/// 数据库存储配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatabaseStorageConfig {
    /// 数据库类型
    pub database_type: DatabaseType,
    /// 连接URL
    pub connection_url: String,
    /// 最大连接数
    pub max_connections: u32,
    /// 最小连接数
    pub min_connections: u32,
    /// 连接超时时间（秒）
    pub connect_timeout_secs: u64,
    /// 空闲连接超时时间（秒）
    pub idle_timeout_secs: u64,
    /// 最大查询超时时间（秒）
    pub max_query_timeout_secs: u64,
    /// 数据表名前缀
    pub table_prefix: String,
    /// 是否启用自动迁移
    pub auto_migrate: bool,
    /// 是否启用连接池监控
    pub enable_pool_monitoring: bool,
    /// 批量操作大小限制
    pub batch_size_limit: usize,
}

impl Default for DatabaseStorageConfig {
    fn default() -> Self {
        Self {
            database_type: DatabaseType::PostgreSQL,
            connection_url: "postgresql://localhost:5432/rustcloud_bus".to_string(),
            max_connections: 10,
            min_connections: 2,
            connect_timeout_secs: 30,
            idle_timeout_secs: 600,
            max_query_timeout_secs: 30,
            table_prefix: "rustcloud_bus_".to_string(),
            auto_migrate: true,
            enable_pool_monitoring: true,
            batch_size_limit: 1000,
        }
    }
}

/// 数据库连接池抽象
#[async_trait]
pub trait DatabasePool: Send + Sync {
    type Connection;
    type Transaction;

    /// 获取数据库连接
    async fn get_connection(&self) -> MessageBusResult<Self::Connection>;
    
    /// 开始事务
    async fn begin_transaction(&self) -> MessageBusResult<Self::Transaction>;
    
    /// 执行查询
    async fn execute_query(
        &self,
        sql: &str,
        params: &[&(dyn sqlx::Encode<sqlx::Postgres> + Send + Sync)]
    ) -> MessageBusResult<u64>;
    
    /// 执行查询并返回结果
    async fn query(
        &self,
        sql: &str,
        params: &[&(dyn sqlx::Encode<sqlx::Postgres> + Send + Sync)]
    ) -> MessageBusResult<Vec<sqlx::postgres::PgRow>>;
    
    /// 健康检查
    async fn health_check(&self) -> MessageBusResult<()>;
    
    /// 获取连接池统计
    async fn get_pool_stats(&self) -> MessageBusResult<PoolStats>;
}

/// 连接池统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolStats {
    /// 总连接数
    pub total_connections: u32,
    /// 活跃连接数
    pub active_connections: u32,
    /// 空闲连接数
    pub idle_connections: u32,
    /// 等待连接数
    pub waiting_connections: u32,
}

/// PostgreSQL连接池实现
pub struct PostgreSQLPool {
    pool: Pool<Postgres>,
    config: DatabaseStorageConfig,
}

impl PostgreSQLPool {
    /// 创建新的PostgreSQL连接池
    pub async fn new(config: DatabaseStorageConfig) -> MessageBusResult<Self> {
        tracing::info!(
            "创建PostgreSQL连接池: url={}, max_conn={}, min_conn={}",
            mask_connection_url(&config.connection_url),
            config.max_connections,
            config.min_connections
        );

        let pool_options = sqlx::postgres::PgPoolOptions::new()
            .max_connections(config.max_connections)
            .min_connections(config.min_connections)
            .connect_timeout(Duration::from_secs(config.connect_timeout_secs))
            .idle_timeout(Duration::from_secs(config.idle_timeout_secs));

        let pool = pool_options
            .connect(&config.connection_url)
            .await
            .map_err(|e| MessageBusError::connection_error(format!("PostgreSQL连接失败: {}", e)))?;

        let instance = Self { pool, config };

        // 执行健康检查
        instance.health_check().await?;
        
        tracing::info!("PostgreSQL连接池创建成功");
        Ok(instance)
    }

    /// 创建数据库表
    pub async fn migrate(&self) -> MessageBusResult<()> {
        tracing::info!("开始执行数据库迁移");

        let messages_table = format!("{}messages", self.config.table_prefix);
        let create_messages_sql = format!(
            r#"
            CREATE TABLE IF NOT EXISTS {} (
                id VARCHAR(36) PRIMARY KEY,
                topic VARCHAR(255) NOT NULL,
                payload JSONB NOT NULL,
                headers JSONB,
                status VARCHAR(20) NOT NULL DEFAULT 'pending',
                retry_count INTEGER NOT NULL DEFAULT 0,
                max_retry_count INTEGER NOT NULL DEFAULT 3,
                stored_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
                processed_at TIMESTAMP WITH TIME ZONE,
                expires_at TIMESTAMP WITH TIME ZONE,
                created_by VARCHAR(100),
                trace_id VARCHAR(64),
                correlation_id VARCHAR(64)
            )
            "#,
            messages_table
        );

        sqlx::query(&create_messages_sql)
            .execute(&self.pool)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("创建messages表失败: {}", e)))?;

        // 创建索引
        let indexes = vec![
            format!("CREATE INDEX IF NOT EXISTS idx_{}_topic ON {} (topic)", 
                self.config.table_prefix.trim_end_matches('_'), messages_table),
            format!("CREATE INDEX IF NOT EXISTS idx_{}_status ON {} (status)", 
                self.config.table_prefix.trim_end_matches('_'), messages_table),
            format!("CREATE INDEX IF NOT EXISTS idx_{}_stored_at ON {} (stored_at)", 
                self.config.table_prefix.trim_end_matches('_'), messages_table),
            format!("CREATE INDEX IF NOT EXISTS idx_{}_expires_at ON {} (expires_at)", 
                self.config.table_prefix.trim_end_matches('_'), messages_table),
            format!("CREATE INDEX IF NOT EXISTS idx_{}_trace_id ON {} (trace_id)", 
                self.config.table_prefix.trim_end_matches('_'), messages_table),
        ];

        for index_sql in indexes {
            sqlx::query(&index_sql)
                .execute(&self.pool)
                .await
                .map_err(|e| MessageBusError::storage_error(format!("创建索引失败: {}", e)))?;
        }

        tracing::info!("数据库迁移完成");
        Ok(())
    }

    /// 获取连接池引用
    pub fn get_pool(&self) -> &Pool<Postgres> {
        &self.pool
    }
}

#[async_trait]
impl DatabasePool for PostgreSQLPool {
    type Connection = sqlx::pool::PoolConnection<Postgres>;
    type Transaction = sqlx::Transaction<'static, Postgres>;

    async fn get_connection(&self) -> MessageBusResult<Self::Connection> {
        self.pool.acquire()
            .await
            .map_err(|e| MessageBusError::connection_error(format!("获取数据库连接失败: {}", e)))
    }

    async fn begin_transaction(&self) -> MessageBusResult<Self::Transaction> {
        self.pool.begin()
            .await
            .map_err(|e| MessageBusError::storage_error(format!("开始事务失败: {}", e)))
    }

    async fn execute_query(
        &self,
        sql: &str,
        _params: &[&(dyn sqlx::Encode<sqlx::Postgres> + Send + Sync)]
    ) -> MessageBusResult<u64> {
        let result = sqlx::query(sql)
            .execute(&self.pool)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("执行查询失败: {}", e)))?;
            
        Ok(result.rows_affected())
    }

    async fn query(
        &self,
        sql: &str,
        _params: &[&(dyn sqlx::Encode<sqlx::Postgres> + Send + Sync)]
    ) -> MessageBusResult<Vec<sqlx::postgres::PgRow>> {
        let rows = sqlx::query(sql)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("查询失败: {}", e)))?;
            
        Ok(rows)
    }

    async fn health_check(&self) -> MessageBusResult<()> {
        sqlx::query("SELECT 1")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| MessageBusError::connection_error(format!("数据库健康检查失败: {}", e)))?;
            
        tracing::debug!("数据库健康检查通过");
        Ok(())
    }

    async fn get_pool_stats(&self) -> MessageBusResult<PoolStats> {
        let pool_state = self.pool.size();
        Ok(PoolStats {
            total_connections: pool_state,
            active_connections: pool_state - self.pool.num_idle(),
            idle_connections: self.pool.num_idle(),
            waiting_connections: 0, // sqlx Pool 没有直接提供这个信息
        })
    }
}

/// 数据库存储实现
pub struct DatabaseStorage<P: DatabasePool> {
    pool: Arc<P>,
    config: DatabaseStorageConfig,
    stats: Arc<RwLock<StorageStats>>,
}

impl<P: DatabasePool> DatabaseStorage<P> {
    /// 创建数据库存储
    pub fn new(pool: P, config: DatabaseStorageConfig) -> Self {
        Self {
            pool: Arc::new(pool),
            config,
            stats: Arc::new(RwLock::new(StorageStats::default())),
        }
    }

    /// 获取表名
    fn get_table_name(&self, table: &str) -> String {
        format!("{}{}", self.config.table_prefix, table)
    }

    /// 构建查询WHERE条件
    fn build_where_clause(&self, criteria: &QueryCriteria) -> (String, Vec<String>) {
        let mut conditions = Vec::new();
        let mut params = Vec::new();
        let mut param_count = 1;

        if let Some(ref status) = criteria.status {
            conditions.push(format!("status = ${}", param_count));
            params.push(status.to_string());
            param_count += 1;
        }

        if let Some(ref topic_pattern) = criteria.topic_pattern {
            if topic_pattern.contains('*') {
                let pattern = topic_pattern.replace('*', '%');
                conditions.push(format!("topic LIKE ${}", param_count));
                params.push(pattern);
            } else {
                conditions.push(format!("topic = ${}", param_count));
                params.push(topic_pattern.clone());
            }
            param_count += 1;
        }

        if let Some(ref start_time) = criteria.start_time {
            conditions.push(format!("stored_at >= ${}", param_count));
            params.push(start_time.to_rfc3339());
            param_count += 1;
        }

        if let Some(ref end_time) = criteria.end_time {
            conditions.push(format!("stored_at <= ${}", param_count));
            params.push(end_time.to_rfc3339());
            param_count += 1;
        }

        let where_clause = if conditions.is_empty() {
            String::new()
        } else {
            format!("WHERE {}", conditions.join(" AND "))
        };

        (where_clause, params)
    }

    /// 构建ORDER BY子句
    fn build_order_clause(&self, order_by: &Option<OrderBy>) -> String {
        if let Some(ref order) = order_by {
            let (column, direction) = match order {
                OrderBy::StoredAt(sort_order) => {
                    let dir = match sort_order {
                        SortOrder::Ascending => "ASC",
                        SortOrder::Descending => "DESC",
                    };
                    ("stored_at", dir)
                }
                OrderBy::Topic(sort_order) => {
                    let dir = match sort_order {
                        SortOrder::Ascending => "ASC",
                        SortOrder::Descending => "DESC",
                    };
                    ("topic", dir)
                }
                OrderBy::Status(sort_order) => {
                    let dir = match sort_order {
                        SortOrder::Ascending => "ASC",
                        SortOrder::Descending => "DESC",
                    };
                    ("status", dir)
                }
            };
            format!("ORDER BY {} {}", column, direction)
        } else {
            "ORDER BY stored_at DESC".to_string()
        }
    }

    /// 将消息记录转换为数据库行
    fn message_record_to_values(record: &MessageRecord) -> MessageBusResult<(
        String, String, serde_json::Value, serde_json::Value, String, i32, i32,
        chrono::DateTime<chrono::Utc>, Option<chrono::DateTime<chrono::Utc>>,
        Option<chrono::DateTime<chrono::Utc>>, Option<String>, Option<String>, Option<String>
    )> {
        let payload = serde_json::to_value(&record.message.payload())
            .map_err(|e| MessageBusError::serialization_error(e.to_string()))?;
            
        let headers = serde_json::to_value(&record.message.headers())
            .map_err(|e| MessageBusError::serialization_error(e.to_string()))?;

        Ok((
            record.message.id().to_string(),
            record.message.topic().to_string(),
            payload,
            headers,
            record.status.to_string(),
            record.retry_count as i32,
            record.max_retry_count as i32,
            record.stored_at,
            record.processed_at,
            record.expires_at,
            record.created_by.clone(),
            record.trace_id.clone(),
            record.correlation_id.clone(),
        ))
    }

    /// 从数据库行创建消息记录
    fn row_to_message_record(&self, row: &sqlx::postgres::PgRow) -> MessageBusResult<MessageRecord> {
        let id: String = row.try_get("id")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取id失败: {}", e)))?;
            
        let topic: String = row.try_get("topic")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取topic失败: {}", e)))?;
            
        let payload: serde_json::Value = row.try_get("payload")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取payload失败: {}", e)))?;
            
        let headers: serde_json::Value = row.try_get("headers")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取headers失败: {}", e)))?;
            
        let headers_map: HashMap<String, String> = serde_json::from_value(headers)
            .map_err(|e| MessageBusError::deserialization_error(format!("解析headers失败: {}", e)))?;

        let mut message = Message::new(&topic, payload);
        message.set_id(&id);
        for (k, v) in headers_map {
            message.set_header(&k, &v);
        }

        let status_str: String = row.try_get("status")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取status失败: {}", e)))?;
            
        let status = status_str.parse()
            .map_err(|e| MessageBusError::deserialization_error(format!("解析status失败: {}", e)))?;

        Ok(MessageRecord {
            message,
            status,
            retry_count: row.try_get::<i32, _>("retry_count").unwrap_or(0) as u32,
            max_retry_count: row.try_get::<i32, _>("max_retry_count").unwrap_or(3) as u32,
            stored_at: row.try_get("stored_at")
                .map_err(|e| MessageBusError::deserialization_error(format!("获取stored_at失败: {}", e)))?,
            processed_at: row.try_get("processed_at").ok(),
            expires_at: row.try_get("expires_at").ok(),
            created_by: row.try_get("created_by").ok(),
            trace_id: row.try_get("trace_id").ok(),
            correlation_id: row.try_get("correlation_id").ok(),
        })
    }
}

/// 数据库事务实现
pub struct DatabaseTransaction<P: DatabasePool> {
    tx: Option<P::Transaction>,
    config: DatabaseStorageConfig,
}

#[async_trait]
impl<P: DatabasePool + 'static> StorageTransaction for DatabaseTransaction<P> {
    async fn store(&mut self, message: Message) -> MessageBusResult<String> {
        let record = MessageRecord::new(message);
        let table_name = format!("{}messages", self.config.table_prefix);
        
        let values = DatabaseStorage::<P>::message_record_to_values(&record)?;
        
        let sql = format!(
            "INSERT INTO {} (id, topic, payload, headers, status, retry_count, max_retry_count, stored_at, processed_at, expires_at, created_by, trace_id, correlation_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
            table_name
        );

        if let Some(ref mut tx) = self.tx {
            sqlx::query(&sql)
                .bind(&values.0)
                .bind(&values.1)
                .bind(&values.2)
                .bind(&values.3)
                .bind(&values.4)
                .bind(&values.5)
                .bind(&values.6)
                .bind(&values.7)
                .bind(&values.8)
                .bind(&values.9)
                .bind(&values.10)
                .bind(&values.11)
                .bind(&values.12)
                .execute(&mut **tx)
                .await
                .map_err(|e| MessageBusError::storage_error(format!("事务存储消息失败: {}", e)))?;

            Ok(record.message.id().to_string())
        } else {
            Err(MessageBusError::storage_error("事务已提交或回滚"))
        }
    }

    async fn commit(mut self: Box<Self>) -> MessageBusResult<()> {
        if let Some(tx) = self.tx.take() {
            tx.commit().await
                .map_err(|e| MessageBusError::storage_error(format!("事务提交失败: {}", e)))?;
            tracing::debug!("数据库事务提交成功");
            Ok(())
        } else {
            Err(MessageBusError::storage_error("事务已提交或回滚"))
        }
    }

    async fn rollback(mut self: Box<Self>) -> MessageBusResult<()> {
        if let Some(tx) = self.tx.take() {
            tx.rollback().await
                .map_err(|e| MessageBusError::storage_error(format!("事务回滚失败: {}", e)))?;
            tracing::debug!("数据库事务回滚成功");
            Ok(())
        } else {
            Err(MessageBusError::storage_error("事务已提交或回滚"))
        }
    }
}

/// 屏蔽连接URL中的敏感信息
fn mask_connection_url(url: &str) -> String {
    if let Some(pos) = url.find("://") {
        if let Some(at_pos) = url[pos + 3..].find('@') {
            let prefix = &url[..pos + 3];
            let suffix = &url[pos + 3 + at_pos + 1..];
            return format!("{}***@{}", prefix, suffix);
        }
    }
    url.to_string()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_mask_connection_url() {
        let url = "postgresql://user:password@localhost:5432/db";
        let masked = mask_connection_url(url);
        assert_eq!(masked, "postgresql://***@localhost:5432/db");
    }

    #[test]
    fn test_database_storage_config_default() {
        let config = DatabaseStorageConfig::default();
        assert_eq!(config.database_type, DatabaseType::PostgreSQL);
        assert_eq!(config.max_connections, 10);
        assert_eq!(config.table_prefix, "rustcloud_bus_");
    }
}

#[async_trait]
impl<P: DatabasePool + 'static> MessageStorage for DatabaseStorage<P> {
    async fn store(&self, message: Message) -> MessageBusResult<String> {
        let record = MessageRecord::new(message);
        let table_name = self.get_table_name("messages");
        
        let values = Self::message_record_to_values(&record)?;
        
        let sql = format!(
            "INSERT INTO {} (id, topic, payload, headers, status, retry_count, max_retry_count, stored_at, processed_at, expires_at, created_by, trace_id, correlation_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
            table_name
        );

        let conn = self.pool.get_connection().await?;
        sqlx::query(&sql)
            .bind(&values.0)
            .bind(&values.1)
            .bind(&values.2)
            .bind(&values.3)
            .bind(&values.4)
            .bind(&values.5)
            .bind(&values.6)
            .bind(&values.7)
            .bind(&values.8)
            .bind(&values.9)
            .bind(&values.10)
            .bind(&values.11)
            .bind(&values.12)
            .execute(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("存储消息失败: {}", e)))?;

        Ok(record.message.id().to_string())
    }

    async fn get(&self, message_id: &str) -> MessageBusResult<Option<MessageRecord>> {
        let table_name = self.get_table_name("messages");
        let sql = format!("SELECT * FROM {} WHERE id = $1", table_name);

        let conn = self.pool.get_connection().await?;
        let row = sqlx::query(&sql)
            .bind(message_id)
            .fetch_optional(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("查询消息失败: {}", e)))?;

        if let Some(row) = row {
            let record = self.row_to_message_record(&row)?;
            Ok(Some(record))
        } else {
            Ok(None)
        }
    }

    async fn delete(&self, message_id: &str) -> MessageBusResult<()> {
        let table_name = self.get_table_name("messages");
        let sql = format!("DELETE FROM {} WHERE id = $1", table_name);

        let conn = self.pool.get_connection().await?;
        sqlx::query(&sql)
            .bind(message_id)
            .execute(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("删除消息失败: {}", e)))?;

        Ok(())
    }

    async fn update_status(&self, message_id: &str, status: MessageStatus) -> MessageBusResult<()> {
        let table_name = self.get_table_name("messages");
        let sql = format!("UPDATE {} SET status = $1 WHERE id = $2", table_name);

        let conn = self.pool.get_connection().await?;
        sqlx::query(&sql)
            .bind(status.to_string())
            .bind(message_id)
            .execute(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("更新消息状态失败: {}", e)))?;

        Ok(())
    }

    async fn update_retry_count(&self, message_id: &str, retry_count: u32) -> MessageBusResult<()> {
        let table_name = self.get_table_name("messages");
        let sql = format!("UPDATE {} SET retry_count = $1 WHERE id = $2", table_name);

        let conn = self.pool.get_connection().await?;
        sqlx::query(&sql)
            .bind(retry_count as i32)
            .bind(message_id)
            .execute(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("更新重试次数失败: {}", e)))?;

        Ok(())
    }

    async fn query_messages(
        &self,
        criteria: QueryCriteria,
        order_by: Option<OrderBy>,
        limit: Option<usize>,
        offset: Option<usize>,
    ) -> MessageBusResult<Vec<MessageRecord>> {
        let table_name = self.get_table_name("messages");
        let (where_clause, params) = self.build_where_clause(&criteria);
        let order_clause = self.build_order_clause(&order_by);

        let mut sql = format!("SELECT * FROM {} {}", table_name, where_clause);
        if let Some(order) = order_clause {
            sql.push_str(&format!(" {}", order));
        }
        if let Some(l) = limit {
            sql.push_str(&format!(" LIMIT {}", l));
        }
        if let Some(o) = offset {
            sql.push_str(&format!(" OFFSET {}", o));
        }

        let conn = self.pool.get_connection().await?;
        let rows = sqlx::query(&sql)
            .bind(&params)
            .fetch_all(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("查询消息失败: {}", e)))?;

        let records = rows
            .into_iter()
            .map(|row| self.row_to_message_record(&row))
            .collect::<MessageBusResult<Vec<MessageRecord>>>()?;

        Ok(records)
    }

    async fn get_stats(&self) -> MessageBusResult<StorageStats> {
        let table_name = self.get_table_name("messages");
        let sql = format!(
            "SELECT COUNT(*) AS total_messages, SUM(CASE WHEN status = 'pending' THEN 1 ELSE 0 END) AS pending_messages, SUM(CASE WHEN status = 'processed' THEN 1 ELSE 0 END) AS processed_messages, SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed_messages FROM {}",
            table_name
        );

        let conn = self.pool.get_connection().await?;
        let row = sqlx::query(&sql)
            .fetch_one(&mut *conn)
            .await
            .map_err(|e| MessageBusError::storage_error(format!("查询统计信息失败: {}", e)))?;

        let total_messages: i64 = row.try_get("total_messages")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取total_messages失败: {}", e)))?;
        let pending_messages: i64 = row.try_get("pending_messages")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取pending_messages失败: {}", e)))?;
        let processed_messages: i64 = row.try_get("processed_messages")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取processed_messages失败: {}", e)))?;
        let failed_messages: i64 = row.try_get("failed_messages")
            .map_err(|e| MessageBusError::deserialization_error(format!("获取failed_messages失败: {}", e)))?;

        Ok(StorageStats {
            total_messages: total_messages as usize,
            pending_messages: pending_messages as usize,
            processed_messages: processed_messages as usize,
            failed_messages: failed_messages as usize,
        })
    }

    async fn begin_transaction(&self) -> MessageBusResult<Box<dyn StorageTransaction>> {
        let tx = self.pool.begin_transaction().await?;
        Ok(Box::new(DatabaseTransaction {
            tx: Some(tx),
            config: self.config.clone(),
        }))
    }

    fn get_retry_delay_strategy(&self) -> RetryDelayStrategy {
        RetryDelayStrategy::ExponentialBackoff {
            initial_delay: Duration::from_secs(1),
            max_delay: Duration::from_secs(60),
            multiplier: 2.0,
        }
    }
}
